/**
 * @file async_mysql.hpp
 * @brief 异步MySQL客户端库，支持同步/异步操作和连接池管理
 * 
 * 该库封装了 MySQL C 客户端 API，提供以下核心功能：
 *   - 同步/异步连接管理
 *   - 同步/异步查询执行
 *   - 结果集处理
 *   - 连接池管理
 *   - 自动重连机制
 *   - 连接健康检查
 * 
 * 特点：
 *   - 基于 ASIO 的高性能异步操作
 *   - 线程安全的连接池实现
 *   - 灵活的 IO 上下文管理（外部或内部）
 *   - 完善的错误处理机制
 * 
 * 使用前配置：
 *   需定义 ASIO 实现宏：
 *   @code
 *   // 使用独立 Asio
 *   #define ASYNC_MYSQL_USE_STANDALONE_ASIO
 *   #include "async_mysql.hpp"
 *   
 *   // 使用 Boost.Asio
 *   #define ASYNC_MYSQL_USE_BOOST_ASIO
 *   #include "async_mysql.hpp"
 *   @endcode
 * 
 * 核心组件：
 *   - async_mysql::connection: 同步连接类
 *   - async_mysql::async_connection: 异步连接类
 *   - async_mysql::connection_pool: 同步连接池
 *   - async_mysql::async_connection_pool: 异步连接池
 * 
 * 
 * @section intro_sec 介绍
 * 本库提供了高效、易用的MySQL数据库访问接口，特别适合需要高并发、异步操作的应用场景。
 * 
 * @section features_sec 主要特性
 * - 同步/异步双模式操作
 * - 自动连接池管理
 * - 连接健康检查与自动重连
 * - 线程安全的API设计
 * - 支持自定义IO上下文
 * 
 * @section usage_sec 使用示例
 * 
 * @subsection async_example 异步使用示例
 * 
 * @code
 * // 使用外部IO上下文
 * asio::io_context io;
 * async_mysql::connection_config config{...};
 * auto pool = async_mysql::async_connection_pool(io, config, 5);
 * 
 * pool.get_connection([](auto conn) {
 *     conn->async_query("SELECT * FROM users", [pool, conn](auto ec, auto rs) {
 *         if (!ec) {
 *             // 处理结果
 *         }
 *         pool.release_connection(conn);
 *     });
 * });
 * 
 * io.run();
 * @endcode
 * 
 * 这里要注意，连接池的连接必须归还，在回调最后需要调用pool.release_connection(conn);
 * 
 * 但async_mmysql也提供了自动归还连接的方案：
 * 
 * @code
 * auto pool = std::make_shared<async_connection_pool>(io, config, 5);
 * 
 * pool->get_connection_proxy([](auto proxy) {
 *     if (!proxy || !*proxy) {
 *         std::cerr << "Failed to get connection\n";
 *         return;
 *     }
 *     
 *     proxy->async_query("SELECT * FROM users", [](auto ec, auto rs) {
 *         if (!ec) {
 *             // 处理结果集
 *         }
 *         // 代理在作用域结束时自动归还连接
 *     });
 * });
 * @endcode
 * 
 * 通过get_connection_proxy获取一个连接代理，代理在生命周期结束时会自动归还连接给连接池
 * 
 * @subsection sync_example 同步使用示例
 * 
 * @code
 * async_mysql::connection_pool sync_pool(config, 5);
 * auto conn = sync_pool.get_connection();
 * auto [ec, rs] = conn->query("SELECT * FROM products");
 * if (!ec) {
 *     // 处理结果
 * }
 * sync_pool.release_connection(conn);
 * @endcode
 * 
 * 同理，同步连接池在最后也要把连接归还给池sync_pool.release_connection(conn);
 * 
 * 同样，同步连接池也提供了自动归还连接的方案：
 * 
 * @code
 * {
 *     // 获取连接代理
 *     auto proxy = pool.get_connection_proxy();
 *     
 *     // 执行查询
 *     auto [ec, rs] = proxy->query("SELECT * FROM users");
 *     
 *     if (!ec) {
 *         // 处理结果集...
 *     }
 *     
 *     // 代理离开作用域时自动归还连接
 * }
 * @endcode
 *  
 * 通过get_connection_proxy获取一个连接代理，代理在生命周期结束时会自动归还连接给连接池
 * 
 * 
 * @copyright MIT License
 * @author czy
 * @version 1.1.0
 * @date 2025-06-15
 */
#ifndef _ASYNC_MYSQL_HPP
#define _ASYNC_MYSQL_HPP

#pragma once

// 检测 ASIO 实现选择
#if defined(ASYNC_MYSQL_USE_BOOST_ASIO) && defined(ASYNC_MYSQL_USE_STANDALONE_ASIO)
    #error "Cannot define both ASYNC_MYSQL_USE_BOOST_ASIO and ASYNC_MYSQL_USE_STANDALONE_ASIO"
#elif !defined(ASYNC_MYSQL_USE_BOOST_ASIO) && !defined(ASYNC_MYSQL_USE_STANDALONE_ASIO)
    // 默认使用独立 ASYNC
    #define ASYNC_MYSQL_USE_STANDALONE_ASIO
#endif

// 包含适当的 ASIO 头文件
#ifdef ASYNC_MYSQL_USE_BOOST_ASIO
    #include <boost/asio.hpp>
    #include <boost/asio/thread_pool.hpp>
    #include <boost/asio/steady_timer.hpp>
    namespace asio = boost::asio;
#else
    #include <asio.hpp>
    #include <asio/thread_pool.hpp>
    #include <asio/steady_timer.hpp>
#endif

#include <mysql/mysql.h>
#include <memory>
#include <queue>
#include <functional>
#include <system_error>
#include <vector>
#include <string>
#include <unordered_map>
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <atomic>
#include <stdexcept>

namespace async_mysql {

// 错误码定义
enum class error {
    success = 0,
    connection_failed,
    query_failed,
    no_more_rows,
    invalid_connection,
    timeout,
    pool_empty
};

// 错误类别
class mysql_error_category : public std::error_category {
public:
    const char* name() const noexcept override { return "mysql"; }
    std::string message(int ev) const override {
        switch (static_cast<error>(ev)) {
            case error::success: return "Success";
            case error::connection_failed: return "Connection failed";
            case error::query_failed: return "Query failed";
            case error::no_more_rows: return "No more rows";
            case error::invalid_connection: return "Invalid connection";
            case error::timeout: return "Operation timed out";
            case error::pool_empty: return "Connection pool empty";
            default: return "Unknown error";
        }
    }
};

inline const std::error_category& get_mysql_error_category() {
    static mysql_error_category instance;
    return instance;
}

inline std::error_code make_error_code(error e) {
    return std::error_code(static_cast<int>(e), get_mysql_error_category());
}

// 结果行封装
class row {
public:
    /**
     * @brief 添加字段到行
     * @param data 字段数据指针
     * @param length 字段数据长度
     */
    void add_field(const char* data, unsigned long length) {
        fields_.emplace_back(data, length);
    }
    
    /**
     * @brief 通过索引访问字段
     * @param index 字段索引
     * @return 字段值引用
     * @throws std::out_of_range 如果索引超出范围
     */
    const std::string& operator[](size_t index) const {
        if (index >= fields_.size()) {
            throw std::out_of_range("Column index out of range");
        }
        return fields_[index];
    }
    
    /**
     * @brief 通过列名访问字段
     * @param name 列名
     * @return 字段值引用
     * @throws std::out_of_range 如果列名不存在
     * 
     * 示例：
     * @code
     *   async_mysql::row r;
     *   // ... 填充行数据 ...
     *   std::cout << "Name: " << r.at("name") << std::endl;
     * @endcode
     */
    const std::string& at(const std::string& name) const {
        if (name_to_index_.empty()) {
            throw std::runtime_error("Field names not initialized");
        }
        auto it = name_to_index_.find(name);
        if (it == name_to_index_.end()) {
            throw std::out_of_range("Column name not found: " + name);
        }
        return fields_[it->second];
    }
    
    /**
     * @brief 获取字段数量
     * @return 字段数量
     */
    size_t size() const noexcept {
        return fields_.size();
    }
    
    /**
     * @brief 设置字段名称映射
     * @param names 字段名称列表
     */
    void set_field_names(const std::vector<std::string>& names) {
        for (size_t i = 0; i < names.size(); ++i) {
            name_to_index_[names[i]] = i;
        }
    }
    
    /**
     * @brief 检查行是否为空
     * @return 如果行为空返回 true，否则返回 false
     */
    bool empty() const noexcept {
        return fields_.empty();
    }
    
private:
    std::vector<std::string> fields_;
    std::unordered_map<std::string, size_t> name_to_index_;
};

// 结果集封装
class result_set {
public:
    result_set() = default;
    
    /**
     * @brief 构造函数
     * @param res MySQL 结果集指针
     */
    explicit result_set(MYSQL_RES* res) : res_(res) {
        if (res_) {
            num_fields_ = mysql_num_fields(res_);
            fields_ = mysql_fetch_fields(res_);
            
            // 初始化字段名
            for (unsigned int i = 0; i < num_fields_; ++i) {
                field_names_.emplace_back(fields_[i].name);
            }
        }
    }
    
    ~result_set() {
        if (res_) {
            mysql_free_result(res_);
        }
    }
    
    // 禁止拷贝
    result_set(const result_set&) = delete;
    result_set& operator=(const result_set&) = delete;
    
    // 允许移动
    result_set(result_set&& other) noexcept 
        : res_(other.res_), fields_(other.fields_), 
          num_fields_(other.num_fields_), field_names_(std::move(other.field_names_)) {
        other.res_ = nullptr;
        other.fields_ = nullptr;
        other.num_fields_ = 0;
    }
    
    result_set& operator=(result_set&& other) noexcept {
        if (this != &other) {
            if (res_) {
                mysql_free_result(res_);
            }
            res_ = other.res_;
            fields_ = other.fields_;
            num_fields_ = other.num_fields_;
            field_names_ = std::move(other.field_names_);
            
            other.res_ = nullptr;
            other.fields_ = nullptr;
            other.num_fields_ = 0;
        }
        return *this;
    }
    
    /**
     * @brief 获取下一行数据
     * @param r 行对象引用，用于存储结果
     * @return 如果成功获取行返回 true，否则返回 false
     * 
     * 示例：
     * @code
     *   async_mysql::result_set rs = ...;
     *   async_mysql::row r;
     *   while (rs.next(r)) {
     *       std::cout << "ID: " << r["id"] << std::endl;
     *   }
     * @endcode
     */
    bool next(row& r) {
        if (!res_) return false;
        
        MYSQL_ROW mysql_row = mysql_fetch_row(res_);
        if (!mysql_row) return false;
        
        unsigned long* lengths = mysql_fetch_lengths(res_);
        if (!lengths) return false;
        
        r = row();
        r.set_field_names(field_names_);
        
        for (unsigned int i = 0; i < num_fields_; ++i) {
            r.add_field(mysql_row[i] ? mysql_row[i] : "", lengths[i]);
        }
        
        return true;
    }
    
    /**
     * @brief 获取行数
     * @return 结果集中的行数
     */
    size_t row_count() const noexcept {
        return res_ ? mysql_num_rows(res_) : 0;
    }
    
    /**
     * @brief 获取字段数量
     * @return 结果集中的字段数量
     */
    size_t field_count() const noexcept {
        return num_fields_;
    }
    
    /**
     * @brief 获取字段名称列表
     * @return 字段名称列表引用
     */
    const std::vector<std::string>& field_names() const noexcept {
        return field_names_;
    }
    
    /**
     * @brief 检查结果集是否为空
     * @return 如果结果集为空返回 true，否则返回 false
     */
    bool empty() const noexcept {
        return row_count() == 0;
    }
    
private:
    MYSQL_RES* res_ = nullptr;
    MYSQL_FIELD* fields_ = nullptr;
    unsigned int num_fields_ = 0;
    std::vector<std::string> field_names_;
};

// 连接配置
struct connection_config {
    std::string host = "localhost";   ///< 数据库主机地址
    std::string user;                 ///< 数据库用户名
    std::string password;             ///< 数据库密码
    std::string database;             ///< 数据库名称
    unsigned int port = 3306;         ///< 数据库端口
    unsigned long client_flag = 0;    ///< MySQL 客户端标志
    std::string unix_socket;          ///< Unix 套接字路径
    unsigned long timeout_sec = 30;   ///< 连接超时时间（秒）
};

// 前置声明
class connection_proxy;
class connection;
class async_connection;
class async_connection_pool;
class connection_pool;

/**
 * @brief MySQL 同步连接类
 * 
 * 此类封装了 MySQL 连接功能，提供同步操作模式：
 *   - 同步方法：connect(), query(), execute() 等
 * 
 * 使用示例：
 * @code
 *   // 同步使用
 *   async_mysql::connection conn(config);
 *   conn.connect();
 *   auto [ec, rs] = conn.query("SELECT * FROM table");
 *   if (!ec) {
 *       // 处理结果
 *   }
 * @endcode
 */
class connection {
public:
    /**
     * @brief 构造函数
     * @param config 数据库连接配置
     */
    explicit connection(const connection_config& config)
        : config_(config) {}
    
    ~connection() {
        close();
    }
    
    // ================= 同步方法 =================
    
    /**
     * @brief 同步连接数据库
     * @return 错误码
     * 
     * 示例：
     * @code
     *   std::error_code ec = conn.connect();
     *   if (!ec) {
     *       // 连接成功
     *   }
     * @endcode
     */
    std::error_code connect() {
        std::lock_guard<std::mutex> lock(mutex_);
        return do_connect();
    }
    
    /**
     * @brief 同步执行查询
     * @param sql SQL 查询语句
     * @return 包含错误码和结果集的 pair
     * 
     * 示例：
     * @code
     *   auto [ec, rs] = conn.query("SELECT * FROM products");
     *   if (!ec) {
     *       // 处理结果集
     *   }
     * @endcode
     */
    std::pair<std::error_code, result_set> query(const std::string& sql) {
        std::lock_guard<std::mutex> lock(mutex_);
        return do_query(sql);
    }
    
    /**
     * @brief 同步执行更新操作
     * @param sql SQL 更新语句（INSERT/UPDATE/DELETE）
     * @return 错误码
     * 
     * 示例：
     * @code
     *   std::error_code ec = conn.execute("UPDATE users SET active = 1");
     *   if (!ec) {
     *       // 更新成功
     *   }
     * @endcode
     */
    std::error_code execute(const std::string& sql) {
        std::lock_guard<std::mutex> lock(mutex_);
        return do_execute(sql);
    }
    
    /**
     * @brief 获取受影响的行数
     * @return 受影响的行数
     */
    uint64_t affected_rows() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return mysql_ ? mysql_affected_rows(mysql_) : 0;
    }
    
    /**
     * @brief 获取最后插入的ID
     * @return 最后插入的ID
     */
    uint64_t last_insert_id() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return mysql_ ? mysql_insert_id(mysql_) : 0;
    }
    
    /**
     * @brief 检查连接是否有效
     * @return 如果连接有效返回 true，否则返回 false
     */
    bool is_connected() const noexcept {
        std::lock_guard<std::mutex> lock(mutex_);
        return mysql_ != nullptr;
    }
    
    /**
     * @brief 关闭数据库连接
     */
    void close() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (mysql_) {
            mysql_close(mysql_);
            mysql_ = nullptr;
        }
    }
    
    /**
     * @brief 重新连接数据库
     * @return 错误码
     * 
     * 如果连接已断开，此方法会尝试重新连接
     */
    std::error_code reconnect() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (mysql_) {
            mysql_close(mysql_);
            mysql_ = nullptr;
        }
        return do_connect();
    }
    
    /**
     * @brief 获取连接的创建时间
     * @return 连接的创建时间点
     */
    std::chrono::steady_clock::time_point creation_time() const noexcept {
        return creation_time_;
    }
    
    // 异常版本方法（可选）
    
    /**
     * @brief 同步连接（异常版本）
     * @throws std::system_error 如果连接失败
     */
    void connect_throw() {
        std::error_code ec = connect();
        if (ec) {
            throw std::system_error(ec, "Database connection failed");
        }
    }
    
    /**
     * @brief 同步查询（异常版本）
     * @param sql SQL 查询语句
     * @return 结果集
     * @throws std::system_error 如果查询失败
     */
    result_set query_throw(const std::string& sql) {
        auto [ec, rs] = query(sql);
        if (ec) {
            throw std::system_error(ec, "Query failed: " + sql);
        }
        return rs;
    }
    
    /**
     * @brief 同步执行（异常版本）
     * @param sql SQL 更新语句
     * @throws std::system_error 如果执行失败
     */
    void execute_throw(const std::string& sql) {
        std::error_code ec = execute(sql);
        if (ec) {
            throw std::system_error(ec, "Execute failed: " + sql);
        }
    }
    
protected:
    // 内部连接实现
    std::error_code do_connect() {
        if (mysql_) {
            return make_error_code(error::success);
        }
        
        mysql_ = mysql_init(nullptr);
        if (!mysql_) {
            return make_error_code(error::connection_failed);
        }
        
        // 设置连接超时
        mysql_options(mysql_, MYSQL_OPT_CONNECT_TIMEOUT, &config_.timeout_sec);
        
        if (!mysql_real_connect(
            mysql_, 
            config_.host.c_str(), 
            config_.user.c_str(), 
            config_.password.c_str(), 
            config_.database.c_str(), 
            config_.port,
            config_.unix_socket.empty() ? nullptr : config_.unix_socket.c_str(),
            config_.client_flag)) {
            
            auto ec = make_error_code(error::connection_failed);
            mysql_close(mysql_);
            mysql_ = nullptr;
            return ec;
        }
        
        creation_time_ = std::chrono::steady_clock::now();
        return make_error_code(error::success);
    }
    
    std::pair<std::error_code, result_set> do_query(const std::string& sql) {
        if (!mysql_) {
            return {make_error_code(error::invalid_connection), result_set()};
        }
        
        if (mysql_real_query(mysql_, sql.c_str(), sql.size())) {
            return {make_error_code(error::query_failed), result_set()};
        }
        
        MYSQL_RES* res = mysql_store_result(mysql_);
        return {make_error_code(error::success), result_set(res)};
    }
    
    std::error_code do_execute(const std::string& sql) {
        if (!mysql_) {
            return make_error_code(error::invalid_connection);
        }
        
        if (mysql_real_query(mysql_, sql.c_str(), sql.size())) {
            return make_error_code(error::query_failed);
        }
        
        return make_error_code(error::success);
    }
protected:
    connection_config config_;
    MYSQL* mysql_ = nullptr;
    mutable std::mutex mutex_;
    std::chrono::steady_clock::time_point creation_time_ = 
        std::chrono::steady_clock::now();
};

/**
 * @brief MySQL 异步连接类
 * 
 * 继承自同步连接类，添加异步操作支持
 * 
 * 使用示例：
 * @code
 *   asio::io_context io;
 *   async_mysql::connection_config config{...};
 *   auto conn = std::make_shared<async_mysql::async_connection>(io, config);
 *   
 *   conn->async_connect([](std::error_code ec) {
 *       if (!ec) {
 *           // 连接成功
 *       }
 *   });
 * @endcode
 */
class async_connection : public connection, public std::enable_shared_from_this<async_connection> {
public:
    using query_callback = std::function<void(std::error_code, result_set)>; ///< 查询回调类型
    using connect_callback = std::function<void(std::error_code)>;          ///< 连接回调类型
    
    /**
     * @brief 构造函数
     * @param io_context ASIO IO 上下文
     * @param config 数据库连接配置
     */
    async_connection(asio::io_context& io_context, 
                    const connection_config& config,
                    std::shared_ptr<asio::thread_pool> thread_pool = nullptr
        ): connection(config), 
          io_context_(io_context),
          thread_pool_(thread_pool ? thread_pool : std::make_shared<asio::thread_pool>(1)) 
    {
    }
    
    ~async_connection() {
        close();
        // 只有当没有外部引用时才join线程池,避免单独使用connection时，connection析构了，回调还在执行
        if (thread_pool_.use_count() == 1) {
            thread_pool_->join();
        }
    }
    
    // ================= 异步方法 =================
    
    /**
     * @brief 异步连接数据库
     * @param callback 连接完成后的回调函数
     * 
     * 示例：
     * @code
     *   conn->async_connect([](std::error_code ec) {
     *       if (!ec) {
     *           std::cout << "Connected successfully\n";
     *       }
     *   });
     * @endcode
     */
    void async_connect(connect_callback callback) {
        asio::post(*thread_pool_, [self = shared_from_this(), callback]() {
            std::error_code ec;
            {
                std::lock_guard<std::mutex> lock(self->mutex_);
                ec = self->do_connect();
            }
            asio::post(self->io_context_, [ec, callback]() { callback(ec); });
        });
    }
    
    /**
     * @brief 异步执行查询
     * @param sql SQL 查询语句
     * @param callback 查询完成后的回调函数
     * 
     * 示例：
     * @code
     *   conn->async_query("SELECT * FROM users", 
     *       [](std::error_code ec, async_mysql::result_set rs) {
     *           if (!ec) {
     *               // 处理结果集
     *           }
     *       });
     * @endcode
     */
    void async_query(const std::string& sql, query_callback callback) {
        asio::post(*thread_pool_, [self = shared_from_this(), sql, callback]() {
            std::pair<std::error_code, result_set> result;
            {
                std::lock_guard<std::mutex> lock(self->mutex_);
                result = self->do_query(sql);
            }
            asio::post(self->io_context_, [result = std::move(result), callback]() mutable { 
                callback(result.first, std::move(result.second)); 
            });
        });
    }
    
    /**
     * @brief 获取线程池
    */
    std::shared_ptr<asio::thread_pool> get_thread_pool() const {
        return thread_pool_;
    }
    
private:
    asio::io_context& io_context_;
    std::shared_ptr<asio::thread_pool> thread_pool_;
};


/**
 * @class async_connection_proxy
 * @brief 异步连接代理类，管理连接生命周期并自动归还
 */
class async_connection_proxy {
public:
    using connection_ptr = std::shared_ptr<async_connection>;
    using pool_ptr = std::shared_ptr<async_connection_pool>;
    
    /**
     * @brief 构造函数
     * @param pool 连接池指针
     * @param conn 连接指针
     * @param auto_release 是否自动归还连接（默认true）
     */
    async_connection_proxy(pool_ptr pool, connection_ptr conn, bool auto_release = true)
        : pool_(std::move(pool)), 
          conn_(std::move(conn)), 
          auto_release_(auto_release) {}
    
    /**
     * @brief 析构函数，自动归还连接
     */
    ~async_connection_proxy() {
        if (auto_release_ && conn_ && pool_) {
            pool_->release_connection(conn_);
        }
    }
    
    // 删除拷贝构造函数和赋值运算符
    async_connection_proxy(const async_connection_proxy&) = delete;
    async_connection_proxy& operator=(const async_connection_proxy&) = delete;
    
    /**
     * @brief 移动构造函数
     */
    async_connection_proxy(async_connection_proxy&& other) noexcept
        : pool_(std::move(other.pool_)),
          conn_(std::move(other.conn_)),
          auto_release_(other.auto_release_) {
        other.conn_ = nullptr;
    }
    
    /**
     * @brief 移动赋值运算符
     */
    async_connection_proxy& operator=(async_connection_proxy&& other) noexcept {
        if (this != &other) {
            release(); // 释放当前资源
            
            pool_ = std::move(other.pool_);
            conn_ = std::move(other.conn_);
            auto_release_ = other.auto_release_;
            
            other.conn_ = nullptr;
        }
        return *this;
    }
    
    /**
     * @brief 手动归还连接
     */
    void release() {
        if (conn_ && pool_) {
            pool_->release_connection(conn_);
            conn_ = nullptr;
        }
    }
    
    /**
     * @brief 禁用自动归还
     */
    void disable_auto_release() noexcept {
        auto_release_ = false;
    }
    
    /**
     * @brief 启用自动归还
     */
    void enable_auto_release() noexcept {
        auto_release_ = true;
    }
    
    /**
     * @brief 获取原始连接指针
     */
    connection_ptr get() const noexcept {
        return conn_;
    }
    
    /**
     * @brief 访问连接方法
     */
    connection_ptr operator->() const noexcept {
        return conn_;
    }
    
    /**
     * @brief 检查代理是否有效
     */
    explicit operator bool() const noexcept {
        return conn_ != nullptr;
    }
    
    /**
     * @brief 执行异步查询（封装方法）
     * @param sql SQL查询语句
     * @param callback 查询完成回调
     */
    template <typename QueryCallback>
    void async_query(const std::string& sql, QueryCallback&& callback) {
        if (!conn_) {
            // 如果连接无效，通过IO上下文报告错误
            asio::post(pool_->get_io_context(), 
                [cb = std::forward<QueryCallback>(callback)]() {
                    cb(make_error_code(error::invalid_connection), result_set());
                });
            return;
        }
        
        conn_->async_query(sql, 
            [self = shared_from_this(), 
             cb = std::forward<QueryCallback>(callback)]
            (auto ec, auto rs) {
                cb(ec, std::move(rs));
            });
    }
    
private:
    pool_ptr pool_;         ///< 连接池指针
    connection_ptr conn_;   ///< 连接指针
    bool auto_release_ = true; ///< 自动归还标志
};

/**
 * @brief 异步连接池
 * 
 * 管理一组数据库连接，支持异步获取和释放连接，包含自动重连功能。
 * 
 * async_connection_pool支持两种构造方法，一种是你的程序已经有asio，或者你想用自己的方式管理io，则使用第一种构造函数
 * @code
 * async_connection_pool(asio::io_context& io, const connection_config& config, 
 *                    size_t pool_size = 5, 
 *                    std::chrono::seconds health_check_interval = std::chrono::seconds(120))
 * @endcode
 * 
 * 这个构造函数需要传入一个io，使用示例：
 * 
 * @code
 *   asio::io_context io;
 *   async_mysql::connection_config config{...};
 *   auto pool = std::make_shared<async_mysql::async_connection_pool>(io, config, 5);
 *   
 *   pool->get_connection([](auto conn) {
 *       conn->async_query("SELECT * FROM table", [pool, conn](auto ec, auto rs) {
 *           if (!ec) {
 *               // 处理结果
 *           }
 *           pool->release_connection(conn);
 *       });
 *   });
 *   
 *   io.run();
 * @endcode
 * 
 * 另外一种构造函数你不需要管asio，直接传入线程数量即可
 * 
 * @note 使用@ref get_connection 方法必须在回调中归还连接，需要执行release_connection
 * 
 * 另外一种方案是使用@ref get_connection_proxy ，这个方法它不需要归还，相对更安全
 * 
 * @code
 * auto pool = std::make_shared<async_connection_pool>(io, config, 5);
 * 
 * pool->get_connection_proxy([](auto proxy) {
 *     if (!proxy || !*proxy) {
 *         std::cerr << "Failed to get connection\n";
 *         return;
 *     }
 *     
 *     proxy->async_query("SELECT * FROM users", [](auto ec, auto rs) {
 *         if (!ec) {
 *             // 处理结果集
 *         }
 *         // 代理在作用域结束时自动归还连接
 *     });
 * });
 * @endcode
 */
class async_connection_pool {
private:
    // 连接+空闲时间记录
    using idle_connection = std::pair<connection_ptr, std::chrono::steady_clock::time_point>;
public:
    using connection_ptr = std::shared_ptr<connection>;
    using get_callback = std::function<void(connection_ptr)>;
    
    /**
     * @brief 构造函数
     * @param io ASIO IO 上下文
     * @param config 数据库连接配置
     * @param thread_pool_size 线程池大小
     * @param initial_size 初始连接数 (默认5)
     * @param min_idle 最小空闲连接数 (默认3)
     * @param max_active 最大活动连接数 (默认20)
     * @param min_evictable_idle_time 最小空闲时间 (默认30分钟)
     * @param health_check_interval 健康检查间隔 (默认30秒)
     */
    async_connection_pool(asio::io_context& io, const connection_config& config, 
                    size_t thread_pool_size = 4,
                    size_t initial_size = 5, 
                    size_t min_idle = 3,
                    size_t max_active = 20,
                    std::chrono::milliseconds min_evictable_idle_time = std::chrono::minutes(30),
                    std::chrono::seconds health_check_interval = std::chrono::seconds(30))
        : io_(io), 
          config_(config), 
          thread_pool_(std::make_shared<asio::thread_pool>(thread_pool_size)),
          min_idle_(min_idle),
          max_active_(max_active),
          min_evictable_idle_time_(min_evictable_idle_time),
          health_check_timer_(io),
          health_check_interval_(health_check_interval) 
    {
        
        // 参数校验
        initial_size = std::min(initial_size, max_active);
        min_idle_ = std::min(min_idle_, max_active);
        // 初始化连接池
        for (size_t i = 0; i < initial_size; ++i) {
            create_and_add_connection();
        }
        
        // 启动健康检查定时器
        start_health_check();
    }
    
     ~async_connection_pool() {
        // 先关闭所有连接
        while (!free_connections_.empty()) {
            auto conn = free_connections_.front();
            conn->close();
            free_connections_.pop();
        }
        
        // 最后释放线程池
        thread_pool_.reset();
    }
    /**
     * @brief 异步获取连接
     * @param callback 获取到连接后的回调
     */
    void get_connection(get_callback callback) {
        asio::post(io_, [this, callback]() {
            if (!free_connections_.empty()) {
                auto entry = free_connections_.front();
                free_connections_.pop();
                auto& conn = entry.first;
                auto& idle_since = entry.second;
                // 验证连接有效性
                if (validate_connection(conn)) {
                    callback(conn);
                } else {
                    handle_invalid_connection(conn, callback);
                }
            } else if (total_connections_ < max_active_) {
                create_and_add_connection(callback);
            } else {
                waiting_callbacks_.push(callback);
            }
        });
    }
    
    /**
     * @brief 释放连接回连接池
     * @param conn 要释放的连接
     */
    void release_connection(connection_ptr conn) {
        asio::post(io_, [this, conn]() {
            if (!validate_connection(conn)) {
                //异常连接，踢掉
                replace_invalid_connection(conn);
                return;
            }
            
            if (!waiting_callbacks_.empty()) {
                auto callback = waiting_callbacks_.front();
                waiting_callbacks_.pop();
                callback(conn);
            } else {
                // 记录空闲起始时间
                free_connections_.emplace(conn, std::chrono::steady_clock::now());
            }
        });
    }
    
        // 验证连接有效性
    bool validate_connection(connection_ptr conn) {
        // 基础检查
        if (!conn || !conn->is_connected()) {
            return false;
        }       
        return true;
    }
private:
    
    
    // 启动健康检查定时器
    void start_health_check() {
        health_check_timer_.expires_after(health_check_interval_);
        health_check_timer_.async_wait([this](std::error_code ec) {
            if (!ec) {
                check_connections_health();
            }
            start_health_check(); // 重新设置定时器
        });
    }
    
    // 连接健康检查（实现动态收缩）
    void check_connections_health() {
        auto now = std::chrono::steady_clock::now();
        size_t valid_count = 0;
        std::queue<idle_connection> valid_connections;

        // 检查空闲连接
        while (!free_connections_.empty()) {
            auto entry = free_connections_.front();
            free_connections_.pop();
            auto& conn = entry.first;
            auto& idle_since = entry.second;

            // 验证连接有效性
            if (!validate_connection(conn)) {
                total_connections_--;
                continue;
            }

            // 检查空闲时间是否超时
            auto idle_duration = now - idle_since;
            if (idle_duration > min_evictable_idle_time_ && 
                total_connections_ > min_idle_) {
                conn->close();
                total_connections_--;
            } else {
                valid_connections.emplace(conn, idle_since);
                valid_count++;
            }
        }
        
        free_connections_ = std::move(valid_connections);
        
        // 补充空闲连接到min_idle
        size_t need_to_create = (min_idle_ > free_connections_.size()) ? 
            (min_idle_ - free_connections_.size()) : 0;
        need_to_create = std::min(need_to_create, max_active_ - total_connections_);
        
        for (size_t i = 0; i < need_to_create; ++i) {
            create_and_add_connection();
        }
    }
    

    
    // 处理无效连接
    void handle_invalid_connection(connection_ptr conn, get_callback callback) {
        // 尝试重连
        try_reconnect(conn, [this, callback](connection_ptr reconnected_conn) {
            if (reconnected_conn && validate_connection(reconnected_conn)) {
                callback(reconnected_conn);
            } else if (total_connections_ < max_active_) {
                create_and_add_connection(callback);
            } else {
                waiting_callbacks_.push(callback);
            }
        });
    }
    
    // 尝试重连
    void try_reconnect(connection_ptr conn, 
                       std::function<void(connection_ptr)> callback,
                       int attempt = 1) {
        asio::post(conn->get_thread_pool(), [this, conn, callback, attempt]() {
            std::error_code ec;
            conn->reconnect();
            
            if (!ec) {
                asio::post(io_, [conn, callback]() {
                    callback(conn);
                });
            } else if (attempt < max_reconnect_attempts_) {
                // 使用指数退避算法重试
                auto delay = std::chrono::milliseconds(200 * (1 << (attempt - 1)));
                
                asio::steady_timer timer(io_, delay);
                timer.async_wait([this, conn, callback, attempt](std::error_code) {
                    try_reconnect(conn, callback, attempt + 1);
                });
            } else {
                // 重连失败
                asio::post(io_, [callback]() {
                    callback(nullptr);
                });
            }
        });
    }
    
    // 替换无效连接
    void replace_invalid_connection(connection_ptr old_conn) {
        // 先移除无效连接
        total_connections_--;
        // 确保关闭无效连接
        if (old_conn) {
            old_conn->close();
        }
        // 尝试创建新连接替换
        if (total_connections_ < max_active_) {
            create_and_add_connection([this](connection_ptr new_conn) {
                if (!waiting_callbacks_.empty()) {
                    auto callback = waiting_callbacks_.front();
                    waiting_callbacks_.pop();
                    callback(new_conn);
                } else {
                    free_connections_.emplace(new_conn, std::chrono::steady_clock::now());
                }
            });
        } else if (!waiting_callbacks_.empty()) {
            // 即使达到最大连接数，如果有等待请求，仍然尝试创建连接
            create_and_add_connection([this](connection_ptr new_conn) {
                if (new_conn) {
                    auto callback = waiting_callbacks_.front();
                    waiting_callbacks_.pop();
                    callback(new_conn);
                } else {
                    // 创建失败时通知等待者
                    auto callback = waiting_callbacks_.front();
                    waiting_callbacks_.pop();
                    callback(nullptr);
                }
            });
        }
    }
    
    // 创建新连接
    void create_and_add_connection(std::function<void(connection_ptr)> callback = nullptr) {
        // 已达上限
        if (total_connections_ >= max_active_) {
            if (callback) callback(nullptr);
            return;
        }

        total_connections_++;
        auto conn = std::make_shared<connection>(io_, config_,thread_pool_);
        
        conn->async_connect([this, conn, callback](std::error_code ec) {
            if (!ec) {
                if (callback) {
                    callback(conn);
                } else {
                    free_connections_.push(conn);
                }
            } else {
                total_connections_--;
                
                // 连接失败，重试或报告错误
                if (callback) {
                    callback(nullptr);
                }
            }
        });
    }
private:
    // 成员变量
    asio::io_context& io_;
    connection_config config_;
    std::shared_ptr<asio::thread_pool> thread_pool_; // 由连接池创建
    
    // 连接存储
    std::queue<idle_connection> free_connections_;
    std::queue<get_callback> waiting_callbacks_;
    std::atomic<size_t> total_connections_ = 0;
    
    // 连接池配置
    const size_t min_idle_;
    const size_t max_active_;
    const std::chrono::milliseconds min_evictable_idle_time_;
    
    // 健康检查
    asio::steady_timer health_check_timer_;
    const std::chrono::seconds health_check_interval_;
};


/**
 * @class connection_proxy
 * @brief 连接代理类，用于自动管理连接归还
 * 
 * 自动归还的例子
 * 
 * @code
 * {
 *    // 获取连接代理
 *    auto proxy = pool.get_connection_proxy();
 *    
 *    // 执行查询
 *    auto [ec, rs] = proxy->query("SELECT * FROM users");
 *    
 *    if (!ec) {
 *        // 处理结果集...
 *    }
 *    
 *    // 代理离开作用域时自动归还连接
 * }
 * @endcode
 * 
 * 你也可以设置为手动归还
 * 
 * @code
 * {
 *    // 获取连接代理并禁用自动归还
 *    auto proxy = pool.get_connection_proxy();
 *    proxy.disable_auto_release();
 *    
 *    try {
 *        // 执行一系列操作...
 *        proxy->execute("BEGIN");
 *        proxy->execute("UPDATE accounts SET balance = balance - 100");
 *        proxy->execute("COMMIT");
 *        
 *        // 手动归还连接
 *        proxy.release();
 *    } catch (...) {
 *        // 发生异常时确保归还连接
 *        if (proxy) proxy.release();
 *        throw;
 *    }
 * }
 * @endcode
 */
class connection_proxy {
public:
    /**
     * @brief 构造函数
     * @param pool 所属连接池
     * @param conn 被代理的连接
     * @param auto_release 是否启用自动归还（默认启用）
     */
    connection_proxy(connection_pool& pool, 
                    std::shared_ptr<connection> conn,
                    bool auto_release = true) noexcept
        : pool_(pool), conn_(std::move(conn)), auto_release_(auto_release) {}
    
    /**
     * @brief 析构函数
     * 
     * 如果启用了自动归还，会自动调用 release() 方法将连接归还给连接池
     */
    ~connection_proxy() {
        if (auto_release_ && conn_) {
            release();
        }
    }
    
    // 禁止拷贝
    connection_proxy(const connection_proxy&) = delete;
    connection_proxy& operator=(const connection_proxy&) = delete;
    
    /**
     * @brief 移动构造函数
     */
    connection_proxy(connection_proxy&& other) noexcept
        : pool_(other.pool_), 
          conn_(std::move(other.conn_)), 
          auto_release_(other.auto_release_) {
        other.conn_ = nullptr;
    }
    
    /**
     * @brief 移动赋值运算符
     */
    connection_proxy& operator=(connection_proxy&& other) noexcept {
        if (this != &other) {
            release(); // 释放当前持有的连接
            
            pool_ = other.pool_;
            conn_ = std::move(other.conn_);
            auto_release_ = other.auto_release_;
            
            other.conn_ = nullptr;
        }
        return *this;
    }
    
    /**
     * @brief 手动归还连接
     * 
     * 提前归还连接并禁用自动归还
     */
    void release() noexcept {
        if (conn_) {
            pool_.release_connection(conn_);
            conn_ = nullptr;
        }
    }
    
    /**
     * @brief 禁用自动归还
     * 
     * 调用此方法后，析构时不会自动归还连接
     * 
     * @warning 禁用自动归还后，必须手动调用 release() 或确保连接被正确归还
     */
    void disable_auto_release() noexcept {
        auto_release_ = false;
    }
    
    /**
     * @brief 启用自动归还
     */
    void enable_auto_release() noexcept {
        auto_release_ = true;
    }
    
    /**
     * @brief 获取原始连接指针
     * @return 原始连接指针
     */
    std::shared_ptr<connection> get() const noexcept {
        return conn_;
    }
    
    /**
     * @brief 访问连接方法
     */
    std::shared_ptr<connection> operator->() const noexcept {
        return conn_;
    }
    
    /**
     * @brief 检查代理是否持有有效连接
     */
    explicit operator bool() const noexcept {
        return conn_ != nullptr;
    }

private:
    connection_pool& pool_;        ///< 所属连接池
    std::shared_ptr<connection> conn_; ///< 被代理的连接
    bool auto_release_ = true;     ///< 自动归还标志
};

/**
 * @brief 同步连接池
 * 
 * 管理一组数据库连接，支持同步获取和释放连接。
 * 
 * 使用示例：
 * @code
 *   async_mysql::connection_config config{...};
 *   async_mysql::connection_pool pool(config, 5);
 *   
 *   auto conn = pool.get_connection();
 *   auto [ec, rs] = conn->query("SELECT * FROM table");
 *   if (!ec) {
 *       // 处理结果
 *   }
 *   pool.release_connection(conn);
 * @endcode
 * 
 * 上面的方法你必须手动调用 release_connection() 方法将连接归还给连接池。
 * 
 * connection_pool提供了另外一种获取连接的方式，即get_connection_proxy()方法
 * 
 * 通过这个方法你获取到的是一个connection_proxy对象，这个对象会自动管理连接的归还。
 * 
 * @code
 * {
 *     // 获取连接代理
 *     auto proxy = pool.get_connection_proxy();
 *     
 *     // 执行查询
 *     auto [ec, rs] = proxy->query("SELECT * FROM users");
 *     
 *     if (!ec) {
 *         // 处理结果集...
 *     }
 *     
 *     // 代理离开作用域时自动归还连接
 * }
 * @endcode
 */
class connection_pool {
public:
    using connection_ptr = std::shared_ptr<connection>; ///< 连接指针类型
    
    /**
     * @brief 构造函数
     * @param config 数据库连接配置
     * @param pool_size 连接池初始大小
     */
    connection_pool(const connection_config& config, size_t pool_size = 5)
        : config_(config),
    {
        for (size_t i = 0; i < pool_size; ++i) {
            free_connections_.push(create_connection());
        }
    }
    
    /**
     * @brief 获取连接（同步）
     * @return 连接指针
     * 
     * 如果没有可用连接，此方法会阻塞直到有连接可用。
     * 
     * @warning 使用此方法必须手动调用 release_connection()
     */
    connection_ptr get_connection() {
        std::unique_lock<std::mutex> lock(mutex_);
        
        while (free_connections_.empty()) {
            if (total_connections_ < max_pool_size_) {
                // 创建新连接
                auto conn = create_connection();
                total_connections_++;
                return conn;
            }
            condition_.wait(lock);
        }
        
        auto conn = free_connections_.front();
        free_connections_.pop();
        
        // 检查连接是否有效
        if (!conn->is_connected()) {
            if (conn->reconnect()) {
                // 重连失败，创建新连接
                conn = create_connection();
            }
        }
        
        return conn;
    }
    
    /**
     * @brief 获取连接代理（推荐使用）
     * @return 连接代理对象
     * 
     * 连接代理会在析构时自动归还连接，确保不会遗漏释放操作。
     */
    connection_proxy get_connection_proxy() {
        return connection_proxy(*this, get_connection());
    }
    
    /**
     * @brief 释放连接回连接池
     * @param conn 要释放的连接
     */
    void release_connection(connection_ptr conn) {
        std::lock_guard<std::mutex> lock(mutex_);
        free_connections_.push(conn);
        condition_.notify_one();
    }
    
private:
    // 创建新连接
    connection_ptr create_connection() {
        auto conn = std::make_shared<connection>(config_);
        std::error_code ec = conn->connect();
        if (ec) {
            throw std::system_error(ec, "Failed to create database connection");
        }
        return conn;
    }
private:
    connection_config config_;
    mutable std::mutex mutex_;
    std::condition_variable condition_;
    std::queue<connection_ptr> free_connections_;
    
    size_t total_connections_ = 0;
    const size_t max_pool_size_ = 20;
};

} // namespace async_mysql

#endif // _ASYNC_MYSQL_HPP